/**
 *  WRMTransactionManager.java        22-ott-2009
 *   
 * Copyright (c) 2009 mentalsmash.org
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 * 
 *  http://www.apache.org/licenses/LICENSE-2.0
 * 
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * 
 *  The use of the Apache License does not indicate that this project is
 *  affiliated with the Apache Software Foundation.
 */

package org.mentalsmash.tazio.transactions.control.impl;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.mentalsmash.tazio.transactions.control.CommitmentAction;
import org.mentalsmash.tazio.transactions.control.CommitmentTask;
import org.mentalsmash.tazio.transactions.control.Snapshooter;
import org.mentalsmash.tazio.transactions.control.SystemTransactionManager;
import org.mentalsmash.tazio.transactions.control.TazioTransaction;
import org.mentalsmash.tazio.transactions.control.TransactionStateException;
import org.mentalsmash.tazio.transactions.control.TransactionsLog;
import org.mentalsmash.tazio.transactions.control.ValidationAgent;
import org.mentalsmash.tazio.transactions.model.RootSet;
import org.mentalsmash.tazio.transactions.model.identifiers.TID;
import org.mentalsmash.tazio.transactions.rm.RMTransactionsFacade;
import org.mentalsmash.tazio.transactions.user.RollbackException;
import org.mentalsmash.tazio.transactions.user.Transaction;
import org.mentalsmash.tazio.transactions.user.TransactionStatus;
import org.mentalsmash.tazio.transactions.user.TransactionalSystemException;
import org.mentalsmash.tazio.transactions.user.UserTransactionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Singleton;

/**
 * The <tt>WRMTransactionManager</tt> models the transaction manager that owns
 * the {@link Transaction} objects in the context of a <i>Weak Replica
 * Manager</i> instance.
 * <p>
 * It provides both system-level operations that are not exposed to end-users
 * and the {@link UserTransactionManager} interface, that let clients interact
 * with the transactional system
 * </p>
 * <p>
 * It implements the <i>singleton</i> pattern, therefore there exists only an
 * instance of <tt>WRMTransactionManager</tt> per <i>Weak Replica Manager </i>
 * </p>
 */
@Singleton
public class WRMTransactionManager implements UserTransactionManager,
	SystemTransactionManager {

    private final static int COMMITTERS_POOL_SIZE = 10;

    private final static Logger log = LoggerFactory
	    .getLogger(WRMTransactionManager.class);

    // instance of TazioTransaction local to a thread
    private final static ThreadLocal<TazioTransaction> currentTransaction = new ThreadLocal<TazioTransaction>();

    private final TransactionsLog transactionsLog;

    private final RMTransactionsFacade replicaManager;

    private final ValidationAgent localValidationAgent;

    private final RootSet rootSet;

    private final Snapshooter snapshooter;

    // TODO: make this injected
    private final ExecutorService executor = Executors
	    .newFixedThreadPool(COMMITTERS_POOL_SIZE);

    private final Lock validationLock = new ReentrantLock();

    private final Lock changesLock = new ReentrantLock();

    @Inject
    WRMTransactionManager(final RMTransactionsFacade rmProxy,
	    final ValidationAgent agent, final TransactionsLog txLog,
	    final RootSet rs, final Snapshooter snapshooter) {
	this.replicaManager = rmProxy;
	this.localValidationAgent = agent;
	this.transactionsLog = txLog;
	this.rootSet = rs;

	this.snapshooter = snapshooter;

    }

    /* ************************************************************************ */
    /* From org.mentalsmash.tazio.transactions.UserTransactionManager interface */
    /* ************************************************************************ */

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transactions.UserTransactionManager#begin()
     */
    @Override
    public Transaction begin() {
	return this.begin(false);

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.UserTransactionManager#beginReadOnly()
     */
    @Override
    public Transaction beginReadOnly() {
	return this.begin(true);
    }

    private Transaction begin(final boolean readonly) {
	TazioTransaction transaction;
	// TODO: Make the following line better
	transaction = (readonly ? new ReadOnlyTransaction(this,
		this.snapshooter, TID.generateTID())
		: new ReadWriteTransaction(this, this.snapshooter, TID
			.generateTID()));

	long initialSeqno;
	currentTransaction.set(transaction);

	transaction.stateTransition(TransactionStatus.BEGINNING);
	this.transactionsLog.logTransaction(transaction);

	initialSeqno = this.replicaManager.beginTransaction(); // remote

	transaction.setSeqNo(initialSeqno);
	transaction.stateTransition(TransactionStatus.RUNNING);

	return transaction;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.UserTransactionManager#getTransaction
     * ()
     */
    @Override
    public Transaction getTransaction() {
	final TazioTransaction tx = currentTransaction.get();
	if (tx == null) {
	    return TazioTransactionImpl.NULL_TRANSACTION;
	} else {
	    return tx;
	}
    }

    /* **************************************************************************
     * 
     * From org.mentalsmash.tazio.transactions.SystemTransactionManager
     * interface
     * 
     * *************************************************************************
     */

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transactions.SystemTransactionManager#commit()
     */
    @Override
    public void commit() throws RollbackException, TransactionStateException,
	    TransactionalSystemException {
	//TODO: when RM is functioning change to WRMCommitmentAction()
	final CommitmentAction action = new WRMCommitmentActionNoRM(
		(TazioTransactionImpl) this.getTransaction(), this,
		this.rootSet);
	final CommitmentTask task = new CommitmentTask(action,this);
	try {
	    this.executor.execute(task);
	    task.get();
	} catch (final InterruptedException e) {
	    log.error("Unexpexted exception while committing transaction {}",
		    getTransaction(), e);
	    throw new TransactionalSystemException(e);
	} catch (final ExecutionException e) {
	    final Throwable cause = e.getCause();
	    log.error("Error while committing transaction {}",
		    getTransaction(), cause);
	    if (cause instanceof RollbackException) {
		throw (RollbackException) cause;
	    } else if (cause instanceof TransactionStateException) {
		throw (TransactionStateException) cause;
	    } else {
		throw new TransactionalSystemException(cause);
	    }
	} 
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.SystemTransactionManager#commitAsync()
     */
    @Override
    public Future<TransactionStatus> commitAsync()
	    throws TransactionStateException {
	//TODO: when RM is functioning change to WRMCommitmentAction()
	final CommitmentAction action = new WRMCommitmentActionNoRM(
		(TazioTransactionImpl) this.getTransaction(), this,
		this.rootSet);
	final CommitmentTask task = new CommitmentTask(action,this);
	this.executor.execute(task);
	return task;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.SystemTransactionManager#getStatus()
     */
    @Override
    public TransactionStatus getStatus() {
	final Transaction tx = getTransaction();
	return tx.getStatus();
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.SystemTransactionManager#
     * getTransactionsLog()
     */
    @Override
    public TransactionsLog getTransactionsLog() {
	return this.transactionsLog;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.SystemTransactionManager#rollback()
     */
    @Override
    public void rollback() throws TransactionStateException {
	final TazioTransactionImpl tx = (TazioTransactionImpl) this
		.getTransaction();
	if (tx.getStatus() != TransactionStatus.RUNNING) {
	    log
		    .error("Transaction"
			    + tx
			    + " that wasn't in the RUNNING status was asked to rollbkack");
	    throw new TransactionStateException("Transaction " + tx
		    + " was not in the RUNNING status");
	}

	tx.stateTransition(TransactionStatus.ROLLED_BACK);
	clearThreadTransaction();
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.SystemTransactionManager#
     * getValidationAgent()
     */
    @Override
    public ValidationAgent getValidationAgent() {
	return this.localValidationAgent;
    }

    /*
     * Other methods
     */
    /**
     * Returns a reference to the local proxy of the <i>Replica Manager</i> used
     * by this <tt>WRMTransactionManager</tt>
     * 
     * @return a reference to the local proxy of the <i>Replica Manager</i> used
     *         by this <tt>WRMTransactionManager</tt>
     * @see RMTransactionsFacade
     */
    protected RMTransactionsFacade getReplicaManagerFacade() {
	return this.replicaManager;
    }

    protected Lock getValidationLock() {
	return this.validationLock;
    }

    protected Lock getChangesLock() {
	return this.changesLock;
    }
    
    public void clearThreadTransaction() {
	WRMTransactionManager.currentTransaction.set(null);
    }

}
